#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
专门调试Celery Worker中的"not enough values to unpack (expected 3, got 0)"错误
重点监控billiard.pool和celery.app.trace模块中的解包操作
"""

import os
import sys
import logging
import traceback
from datetime import datetime
import threading
import time
import importlib

# 设置详细日志，确保所有信息都被记录
logging.basicConfig(
    level=logging.DEBUG,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler("celery_unpack_error_debug.log"),
        logging.StreamHandler()
    ]
)
logger = logging.getLogger(__name__)

# 添加异常钩子，捕获所有未处理的异常
def exception_hook(exc_type, exc_value, exc_traceback):
    if issubclass(exc_type, KeyboardInterrupt):
        sys.__excepthook__(exc_type, exc_value, exc_traceback)
        return
    
    logger.error("未捕获的异常:", exc_info=(exc_type, exc_value, exc_traceback))

sys.excepthook = exception_hook

# 提前导入可能的问题模块，尝试捕获导入时的问题
try:
    import billiard
    logger.info(f"已导入billiard模块，版本: {getattr(billiard, '__version__', '未知')}")
except Exception as e:
    logger.error(f"导入billiard模块时出错: {str(e)}")
    logger.error(traceback.format_exc())

try:
    import celery
    logger.info(f"已导入celery模块，版本: {getattr(celery, '__version__', '未知')}")
except Exception as e:
    logger.error(f"导入celery模块时出错: {str(e)}")
    logger.error(traceback.format_exc())

# 设置环境变量以启用调试模式
os.environ['TASK_DEBUG_MODE'] = 'True'
# 禁用Celery的eager模式，强制使用异步模式
os.environ['CELERY_TASK_ALWAYS_EAGER'] = 'False'

# 设置Django环境
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'baojing.settings')

# 导入Django并初始化
try:
    import django
    django.setup()
    logger.info(f"Django已初始化，版本: {django.get_version()}")
except Exception as e:
    logger.error(f"Django初始化失败: {str(e)}")
    logger.error(traceback.format_exc())
    sys.exit(1)

# 导入Celery应用
try:
    from baojing.celery import app
    logger.info("Celery应用已导入")
except Exception as e:
    logger.error(f"导入Celery应用失败: {str(e)}")
    logger.error(traceback.format_exc())
    sys.exit(1)

# 导入相关任务和模型
try:
    from tasks.task_scheduler import check_and_schedule_tasks, execute_scheduled_task
    from tasks.tasks import collect_and_push_pending_items
    from tasks.models import ScheduledTask, TaskExecutionLog
    from projects.models import Project, Message
    logger.info("已导入所有必要的任务和模型")
except Exception as e:
    logger.error(f"导入任务和模型时出错: {str(e)}")
    logger.error(traceback.format_exc())

# 猴子补丁：为billiard.pool.Worker.apply_async添加包装器
def monkey_patch_billiard_pool():
    """对billiard.pool模块进行猴子补丁，监控可能的解包错误"""
    try:
        import billiard.pool
        original_apply_async = billiard.pool.Worker.apply_async
        
        def patched_apply_async(self, func, args=(), kwds={}, callback=None, error_callback=None, **options):
            logger.debug(f"调用billiard.pool.Worker.apply_async: func={func.__name__ if hasattr(func, '__name__') else str(func)}")
            logger.debug(f"参数: args={args}, kwds={kwds}")
            
            # 包装原始回调函数，捕获可能的错误
            if callback:
                original_callback = callback
                
                def wrapped_callback(result):
                    try:
                        logger.debug(f"执行回调函数，结果类型: {type(result)}")
                        logger.debug(f"结果内容: {result}")
                        return original_callback(result)
                    except ValueError as e:
                        if "not enough values to unpack" in str(e):
                            logger.error(f"捕获到目标解包错误在回调中: {str(e)}")
                            logger.error(f"尝试解包的结果: {result}")
                            logger.error(traceback.format_exc())
                        raise
                
                callback = wrapped_callback
            
            try:
                return original_apply_async(self, func, args, kwds, callback, error_callback, **options)
            except ValueError as e:
                if "not enough values to unpack" in str(e):
                    logger.error(f"捕获到目标解包错误在apply_async中: {str(e)}")
                    logger.error(f"函数: {func}, 参数: {args}, {kwds}")
                    logger.error(traceback.format_exc())
                raise
        
        # 应用补丁
        billiard.pool.Worker.apply_async = patched_apply_async
        logger.info("已成功为billiard.pool.Worker.apply_async应用猴子补丁")
        
    except Exception as e:
        logger.error(f"应用billiard.pool猴子补丁时出错: {str(e)}")
        logger.error(traceback.format_exc())

# 猴子补丁：为celery.app.trace.fast_trace_task添加包装器
def monkey_patch_celery_trace():
    """对celery.app.trace模块进行猴子补丁，直接监控fast_trace_task函数"""
    try:
        from celery.app import trace
        original_fast_trace_task = trace.fast_trace_task
        
        def patched_fast_trace_task(task_id, args, kwargs, request=None):
            logger.debug(f"调用celery.app.trace.fast_trace_task: task_id={task_id}")
            logger.debug(f"参数: args={args}, kwargs={kwargs}")
            
            try:
                # 尝试捕获_loc变量的内容
                if hasattr(trace, '_loc'):
                    logger.debug(f"当前trace._loc的值: {trace._loc}")
                    if trace._loc is None or len(trace._loc) != 3:
                        logger.warning(f"发现异常的_loc值: {trace._loc}")
                
                result = original_fast_trace_task(task_id, args, kwargs, request)
                return result
            except ValueError as e:
                if "not enough values to unpack" in str(e):
                    logger.error(f"捕获到目标解包错误在fast_trace_task中: {str(e)}")
                    # 尝试获取当前的_loc值
                    loc_value = getattr(trace, '_loc', '未找到')
                    logger.error(f"当前trace._loc的值: {loc_value}")
                    logger.error(f"任务ID: {task_id}, 参数: {args}, {kwargs}")
                    logger.error(traceback.format_exc())
                raise
        
        # 应用补丁
        trace.fast_trace_task = patched_fast_trace_task
        logger.info("已成功为celery.app.trace.fast_trace_task应用猴子补丁")
        
    except Exception as e:
        logger.error(f"应用celery.app.trace猴子补丁时出错: {str(e)}")
        logger.error(traceback.format_exc())

# 创建一个模拟任务，故意触发不同类型的错误
def create_test_task():
    """创建一个测试任务用于调试"""
    try:
        # 检查是否已经存在测试任务
        existing_task = ScheduledTask.objects.filter(name="解包错误测试任务").first()
        if existing_task:
            logger.info(f"测试任务已存在: {existing_task.id}")
            return existing_task
            
        # 创建新的测试任务
        test_task = ScheduledTask.objects.create(
            name="解包错误测试任务",
            description="用于调试解包错误的测试任务",
            cron_expression="* * * * *",  # 每分钟执行一次
            is_active=True
        )
        logger.info(f"已创建测试任务: {test_task.id}")
        return test_task
    except Exception as e:
        logger.error(f"创建测试任务时出错: {str(e)}")
        logger.error(traceback.format_exc())
        return None

# 测试函数：尝试以不同方式执行任务以触发错误
def test_task_execution():
    """测试不同的任务执行方式"""
    logger.info("=== 开始测试任务执行 ===")
    
    # 测试1: 直接调用函数
    try:
        logger.info("测试1: 直接调用check_and_schedule_tasks函数")
        result = check_and_schedule_tasks()
        logger.info(f"直接调用结果: {result}")
    except Exception as e:
        logger.error(f"直接调用任务时出错: {str(e)}")
        logger.error(traceback.format_exc())
    
    # 测试2: 使用delay异步执行
    try:
        logger.info("测试2: 使用delay异步执行任务")
        result = check_and_schedule_tasks.delay()
        logger.info(f"异步任务ID: {result.id}")
        # 等待任务完成
        time.sleep(5)
        if result.ready():
            try:
                task_result = result.get(timeout=1)
                logger.info(f"异步任务结果: {task_result}")
            except Exception as e:
                logger.error(f"获取异步任务结果时出错: {str(e)}")
        else:
            logger.warning("异步任务尚未完成")
    except Exception as e:
        logger.error(f"使用delay执行任务时出错: {str(e)}")
        logger.error(traceback.format_exc())
    
    # 测试3: 直接创建并执行一个测试任务
    test_task = create_test_task()
    if test_task:
        try:
            logger.info(f"测试3: 执行测试任务 {test_task.id}")
            result = execute_scheduled_task.delay(test_task.id)
            logger.info(f"测试任务ID: {result.id}")
            time.sleep(5)
        except Exception as e:
            logger.error(f"执行测试任务时出错: {str(e)}")
            logger.error(traceback.format_exc())
    
    # 测试4: 执行collect_and_push_pending_items任务
    try:
        logger.info("测试4: 执行collect_and_push_pending_items任务")
        result = collect_and_push_pending_items.delay()
        logger.info(f"推送任务ID: {result.id}")
        time.sleep(5)
    except Exception as e:
        logger.error(f"执行推送任务时出错: {str(e)}")
        logger.error(traceback.format_exc())
    
    logger.info("=== 任务执行测试完成 ===")

# 监控函数：持续监控Celery状态
def monitor_celery_status():
    """持续监控Celery状态"""
    logger.info("=== 开始监控Celery状态 ===")
    
    # 获取当前Beat调度配置
    try:
        beat_schedule = app.conf.beat_schedule
        logger.info(f"当前Beat调度配置: {len(beat_schedule)}个任务")
        for key, task_config in beat_schedule.items():
            logger.debug(f"调度任务 {key}: {task_config}")
    except Exception as e:
        logger.error(f"获取Beat调度配置时出错: {str(e)}")
    
    # 获取当前任务执行模式
    try:
        is_eager = app.conf.task_always_eager
        logger.info(f"当前任务执行模式: {'同步模式' if is_eager else '异步模式'}")
    except Exception as e:
        logger.error(f"获取任务执行模式时出错: {str(e)}")
    
    # 检查Redis连接状态
    try:
        from celery import current_app
        inspector = current_app.control.inspect()
        workers = inspector.ping()
        if workers:
            logger.info(f"找到{len(workers)}个活跃的Celery Worker")
        else:
            logger.warning("未找到活跃的Celery Worker")
    except Exception as e:
        logger.error(f"检查Worker状态时出错: {str(e)}")
        logger.debug("可能是因为Redis连接问题或Worker未运行")
    
    logger.info("=== Celery状态监控完成 ===")

# 主函数
def main():
    """主函数"""
    logger.info("=== 开始调试 'not enough values to unpack (expected 3, got 0)' 错误 ===")
    logger.info(f"当前时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
    
    # 应用猴子补丁
    monkey_patch_billiard_pool()
    monkey_patch_celery_trace()
    
    # 监控当前Celery状态
    monitor_celery_status()
    
    # 运行任务执行测试
    test_task_execution()
    
    # 等待一段时间，观察是否有后台错误发生
    logger.info("等待60秒观察是否有后台错误...")
    time.sleep(60)
    
    # 再次运行测试以增加触发错误的机会
    test_task_execution()
    
    logger.info("=== 调试完成 ===")
    logger.info(f"请查看 celery_unpack_error_debug.log 获取详细信息")

if __name__ == "__main__":
    try:
        main()
    except Exception as e:
        logger.error(f"主程序执行失败: {str(e)}")
        logger.error(traceback.format_exc())
        sys.exit(1)